package demo.mq.rabbit;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;

import demo.java.lang.ThreadDemo;

public class ProducerDemo {

    private static Logger logger = LoggerFactory.getLogger(ProducerDemo.class);




    /**
     * 
     * <li>* (star) can substitute for exactly one word.
     * <li># (hash) can substitute for zero or more words.
     * <p>
     * Topic exchange is powerful and can behave like other exchanges.
     * 
     * When a queue is bound with "#" (hash) binding key - it will receive all the
     * messages, regardless of the routing key - like in fanout exchange.
     * 
     * When special characters, "*" (star) and "#" (hash), aren't used in bindings,
     * the topic exchange will behave just like a direct one.
     */
    @Test
    public void sendMsgByTopicExchange() {
        String exchangeName = "topic_logs";
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
            // 声明topic exchange
            channel.exchangeDeclare(exchangeName, "topic");
            for (int i = 0; i < 100; i++) {
                String message = i + " Hello World!";

                String routingKey = String.format("%s.%s", i % 2, i % 5);

                channel.basicPublish(exchangeName, routingKey, null, message.getBytes("UTF-8"));
                System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
                ThreadDemo.safeSleep(500);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }

    /**
     * <link>https://github.com/rabbitmq/rabbitmq-tutorials/blob/master/java/RPCClient.java
     */
    @Test
    public void rpcClient() {
        String requestQueueName = "rpc_queue";
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {

            String corrId = UUID.randomUUID().toString();

            String replyQueueName = channel.queueDeclare().getQueue();

            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId)
                    .replyTo(replyQueueName).build();

            String message = "10";
            channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

            final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);

            String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
                if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                    response.offer(new String(delivery.getBody(), "UTF-8"));
                }
            }, consumerTag -> {
            });

            String result = response.take();
            channel.basicCancel(ctag);
            System.out.println(result);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
